package com.futu.attraction.config;


import org.apache.flink.api.common.functions.AggregateFunction;

/**
 * Flink聚合函数：计算温度平均值
 */
public class AverageAggregateFunction implements AggregateFunction<Double, AverageAccumulator, Double> {

    /**
     * 创建累加器
     */
    @Override
    public AverageAccumulator createAccumulator() {
        return new AverageAccumulator(0.0, 0);
    }

    /**
     * 将元素添加到累加器
     */
    @Override
    public AverageAccumulator add(Double value, AverageAccumulator accumulator) {
        return new AverageAccumulator(
                accumulator.getSum() + value,
                accumulator.getCount() + 1
        );
    }

    /**
     * 获取聚合结果
     */
    @Override
    public Double getResult(AverageAccumulator accumulator) {
        if (accumulator.getCount() == 0) {
            return 0.0;
        }
        return accumulator.getSum() / accumulator.getCount();
    }

    /**
     * 合并两个累加器
     */
    @Override
    public AverageAccumulator merge(AverageAccumulator a, AverageAccumulator b) {
        return new AverageAccumulator(
                a.getSum() + b.getSum(),
                a.getCount() + b.getCount()
        );
    }
}

